package com.lrd.rmq514version.rmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

@Component
public class MyDelayTimeMsgConsumer implements MessageListenerConcurrently {

    private static final Logger logger = LoggerFactory.getLogger(MyDelayTimeMsgConsumer.class);
    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    /**
     * 自产自消消费组
     */
    private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();

    @Value("${apache.rocketmq.namesrvAddr}")
    private String nameServer;

    @Value("${apache.rocketmq.delay-time-consumer.topic}")
    private String delayTimeTopic;

    @Value("${apache.rocketmq.delay-time-consumer.group}")
    private String group;


    /**
     * 初始化
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void start() {
        try {
            logger.info("MyDelayTimeMsgConsumer|starting....");
            consumer.setNamesrvAddr(nameServer);
            consumer.setConsumerGroup(group);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            // 集群消费模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            // 订阅主题
            consumer.subscribe(delayTimeTopic, "*");
            consumer.setPullThresholdForQueue(5);
            consumer.setPullThresholdForTopic(5);
            // 注册消息监听器
            consumer.registerMessageListener(this);
            // 启动消费端
            consumer.start();
            logger.info("MyDelayTimeMsgConsumer|消费者启动成功");
        } catch (MQClientException e) {
            logger.error("MyDelayTimeMsgConsumer|消费者启动异常|errorCode={}, errorMsg = {}", e.getResponseCode(), e.getErrorMessage(), e);
        }

    }


    /**
     * 消费消息
     *
     * @param msgs
     * @param context
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        MessageExt msg = null;
        try {
            for (MessageExt messageExt : msgs) {
                msg = messageExt;
                printLog(msg);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            logger.error("MyDelayTimeMsgConsumer|消费异常,msgKey={}", msg.getKeys(), e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    /**
     * 打印日志
     *
     * @param msg
     */
    private void printLog(MessageExt msg) throws Exception {
        String messageBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
        String msgId = msg.getMsgId();
        String keys = msg.getKeys();
        String currentTime = LocalDateTime.now().format(formatter);
        logger.info("延时消息[消费]成功,消费时间={},msgKey={},msgId={},消息内容={}", currentTime, keys, msgId, messageBody);
    }


    @PreDestroy
    public void stop() {
        consumer.shutdown();
        logger.warn("MyDelayTimeMsgConsumer|stop consumer!");
    }

}
